package com.atguigu.tingshu.search.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.extra.pinyin.PinyinUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.LongTermsBucket;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.CompletionSuggestOption;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.HitsMetadata;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import co.elastic.clients.json.JsonData;
import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.AlbumFeignClient;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.model.album.*;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.repository.AlbumInfoIndexRepository;
import com.atguigu.tingshu.search.repository.SuggestIndexRepository;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserFeignClient;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;


@Slf4j
@Service
@SuppressWarnings({"all"})

public class SearchServiceImpl implements SearchService {

    //要封装的这个类需要两个远程调用的接口和其中的三个方法,由于专辑细节表的远程调用有点困难所以目前先随机生成模仿远程调用的结果
    //todo 后续将专辑细节表的信息改为远程调用的结果

    @Autowired
    private UserFeignClient userFeignClient;

    @Autowired
    private AlbumFeignClient albumFeignClient;

    @Autowired
    private AlbumInfoIndexRepository albumInfoIndexRepository;

    @Autowired
    private Executor threadPoolTaskExecutor;

    @Autowired
    private ElasticsearchClient elasticsearchClient;

    @Autowired
    private SuggestIndexRepository suggestIndexRepository;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;


    /**
     * 将指定专辑上架到索引库
     *
     * @param albumId 专辑ID
     * @return
     */
    @Override
    public void upperAlbum(Long albumId) {
        //1.创建索引库文档对象 AlbumInfoIndex
        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();

        //创建异步编程的代码
        CompletableFuture<AlbumInfo> albumInfoCompletableFuture = CompletableFuture.supplyAsync(() -> {

            //2.先封装albumInfo的信息,由远程调用获取
            AlbumInfo albumInfo = albumFeignClient.getAlbumInfo(albumId).getData();

            if (albumInfo == null) {
                log.error("专辑:" + albumId + "不存在");
                throw new RuntimeException("专辑不存在");
            }


            //封装albumInfoIndex的一部分信息
            BeanUtil.copyProperties(albumInfo, albumInfoIndex);

            //封装属性值集合List<AlbumAttributeValue> 到 albumInfoIndex 中
            List<AlbumAttributeValue> albumAttributeValueVoList = albumInfo.getAlbumAttributeValueVoList();
            if (CollUtil.isNotEmpty(albumAttributeValueVoList)) {
                List<AttributeValueIndex> attributeValueIndexList = albumAttributeValueVoList
                        .stream()
                        .map(value -> {
                            return BeanUtil.copyProperties(value, AttributeValueIndex.class);
                        }).collect(Collectors.toList());

                albumInfoIndex.setAttributeValueIndexList(attributeValueIndexList);
            }
            return albumInfo;

        }, threadPoolTaskExecutor);


        //调用albumFeignClient远程接口,封装分类信息
        CompletableFuture<Void> voidCompletableFuture2 = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            BaseCategoryView baseCategoryView = albumFeignClient.getCategoryView(albumInfo.getCategory3Id()).getData();
            if (baseCategoryView == null) {
                log.error("分类:" + baseCategoryView + "不存在");
                throw new RuntimeException("分类不存在");
            }

            //封装分类信息,注意,由于两张表不是一个主键,所以无法使用beanutil
            albumInfoIndex.setCategory1Id(baseCategoryView.getCategory1Id());
            albumInfoIndex.setCategory2Id(baseCategoryView.getCategory2Id());

        }, threadPoolTaskExecutor);

        //调用用户信息openfeign接口
        CompletableFuture<Void> voidCompletableFuture1 = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            UserInfoVo userInfoVo = userFeignClient.getUserInfoVo(albumInfo.getUserId()).getData();

            if (userInfoVo == null) {
                log.error("用户:" + albumInfo.getUserId() + "不存在");
                throw new RuntimeException("用户不存在");
            }
            albumInfoIndex.setAnnouncerName(userInfoVo.getNickname());
        }, threadPoolTaskExecutor);


        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(() -> {
            //5.封装文档对象中统计相关信息 TODO 采用随机生成方式
            //5.1 封装播放量数值
            int playStatNum = RandomUtil.randomInt(1000, 2000);
            albumInfoIndex.setPlayStatNum(playStatNum);
            //5.2 封装订阅量数值
            int subscribeStatNum = RandomUtil.randomInt(800, 1000);
            albumInfoIndex.setSubscribeStatNum(subscribeStatNum);
            //5.3 封装购买量数值
            int buyStatNum = RandomUtil.randomInt(100, 500);
            albumInfoIndex.setBuyStatNum(buyStatNum);
            //5.4 封装评论量数值
            int commentStatNum = RandomUtil.randomInt(500, 1000);
            albumInfoIndex.setCommentStatNum(commentStatNum);
            //5.5 基于以上生成统计数值计算出当前文档热度分值  热度=累加（不同统计数值*权重）
            BigDecimal bigDecimal1 = new BigDecimal("0.1").multiply(BigDecimal.valueOf(playStatNum));
            BigDecimal bigDecimal2 = new BigDecimal("0.2").multiply(BigDecimal.valueOf(subscribeStatNum));
            BigDecimal bigDecimal3 = new BigDecimal("0.3").multiply(BigDecimal.valueOf(buyStatNum));
            BigDecimal bigDecimal4 = new BigDecimal("0.4").multiply(BigDecimal.valueOf(commentStatNum));
            BigDecimal hotScore = bigDecimal1.add(bigDecimal2).add(bigDecimal3).add(bigDecimal4);
            albumInfoIndex.setHotScore(hotScore.doubleValue());

        }, threadPoolTaskExecutor);

        CompletableFuture.allOf(
                albumInfoCompletableFuture,
                voidCompletableFuture2,
                voidCompletableFuture1,
                voidCompletableFuture
        ).join();

        //6.保存专辑索引库文档对象
        albumInfoIndexRepository.save(albumInfoIndex);

        //将专辑标题存入提词索引库
        this.saveSuggestIndex(albumInfoIndex);

        //将上架的专辑id存入布隆过滤器
        RBloomFilter<Object> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);

        bloomFilter.add(albumId);
    }


    /**
     * 下架专辑-删除文档
     *
     * @param albumId
     * @return
     */
    @Override
    public void lowerAlbum(Long albumId) {
        albumInfoIndexRepository.deleteById(albumId);
    }

    @Override
    public AlbumSearchResponseVo searchAlbum(AlbumIndexQuery albumIndexQuery) {

        //1.构建完整的请求对象
        SearchRequest searchRequest = this.buildDSL(albumIndexQuery);


        System.out.println();
        log.info("本次检索的DSL:");
        System.err.println(searchRequest.toString());
        SearchResponse<AlbumInfoIndex> searchResponse = null;
        try {
            //2.执行检索
            searchResponse = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);

            //3.解析结果
            AlbumSearchResponseVo albumSearchResponseVo = this.parseResult(searchResponse, albumIndexQuery);

            return albumSearchResponseVo;
        } catch (IOException e) {
            log.error("es解析失败");
            throw new RuntimeException(e);
        }


    }

    private static final String INDEX_NAME = "albuminfo";

    /**
     * 构建完整的检索请求对象,包含所有的DSL请求体参数
     *
     * @param albumIndexQuery
     * @return
     */
    @Override
    public SearchRequest buildDSL(AlbumIndexQuery albumIndexQuery) {

        SearchRequest.Builder searchRequestBuilder = new SearchRequest.Builder();

        searchRequestBuilder.index(INDEX_NAME);

        Integer from = (albumIndexQuery.getPageNo() - 1) * albumIndexQuery.getPageSize();
        searchRequestBuilder.from(from).size(albumIndexQuery.getPageSize());

        String keyword = albumIndexQuery.getKeyword();

        if (StringUtils.isNotBlank(keyword)) {
            searchRequestBuilder.highlight(h -> h.fields("albumTitle", t -> t.preTags("<font style = 'color:red'>").postTags("</font>")));
        }


        String order = albumIndexQuery.getOrder();
        if (StringUtils.isNotBlank(order)) {
            String[] split = order.split(":");
            if (split != null && split.length == 2) {

                String orderField = "";
                switch (split[0]) {
                    case "1":
                        orderField = "hotScore";
                        break;
                    case "2":
                        orderField = "playStatNum";
                        break;
                    case "3":
                        orderField = "createTime";
                        break;
                }

                SortOrder sortOrder = split[1].equals("asc") ? SortOrder.Asc : SortOrder.Desc;
                String finalOrderField = orderField;
                searchRequestBuilder.sort(s -> s.field(f -> f.field(finalOrderField).order(sortOrder)));


            }
        }

        //创建bool查询组合三大查询条件
        //1. 处理关键字查询(must)

        BoolQuery.Builder boolQueryBuilder = new BoolQuery.Builder();

        if (StringUtils.isNotBlank(keyword)) {
            boolQueryBuilder.must(s -> s.match(f -> f.field("albumTitle").query(keyword)));
        }

        //2. 处理1,2,3级分类

        Long category1Id = albumIndexQuery.getCategory1Id();
        if (category1Id != null) {
            boolQueryBuilder.filter(f -> f.term(t -> t.field("category1Id").value(category1Id)));
        }
        Long category2Id = albumIndexQuery.getCategory2Id();
        if (category2Id != null) {
            boolQueryBuilder.filter(f -> f.term(t -> t.field("category2Id").value(category2Id)));

        }
        Long category3Id = albumIndexQuery.getCategory3Id();
        if (category3Id != null) {
            boolQueryBuilder.filter(f -> f.term(t -> t.field("category3Id").value(category3Id)));
            //boolQueryBuilder.filter()
        }
        //3. 处理多组标签过滤条件
        List<String> attributeList = albumIndexQuery.getAttributeList();
        if (CollUtil.isNotEmpty(attributeList)) {
            for (String attribute : attributeList) {
                //没遍历一次,就封装一次nested查询
                String[] attributeSplit = attribute.split(":");
                if (attributeSplit != null && attributeSplit.length == 2) {
                    String attributeId = attributeSplit[0];
                    String attributeValueId = attributeSplit[1];
                    boolQueryBuilder.filter(s -> s.nested(f -> f.path("attributeValueIndexList").query(m -> m.bool(b -> b.must(q -> q.term(t -> t.field("attributeValueIndexList.attributeId").value(attributeId)))
                            .must(q -> q.term(t -> t.field("attributeValueIndexList.valueId").value(attributeValueId)))))));

                }


            }
        }

        searchRequestBuilder.query(boolQueryBuilder.build()._toQuery());

        //searchRequestBuilder.query();

        searchRequestBuilder.source(s -> s.filter(f -> f.excludes("attributeValueIndexList",
                "hotScore",
                "commentStatNum",
                "buyStatNum",
                "subscribeStatNum",
                "announcerName")));

        return searchRequestBuilder.build();
    }

    /**
     * 解析ES返回的检索结果
     *
     * @param searchResponse
     * @param albumIndexQuery
     * @return
     */
    @Override
    public AlbumSearchResponseVo parseResult(SearchResponse<AlbumInfoIndex> searchResponse, AlbumIndexQuery albumIndexQuery) {

        //1.创建响应VO对象
        AlbumSearchResponseVo albumSearchResponseVo = new AlbumSearchResponseVo();

        //2.封装VO中分页相关四项属性
        Integer pageNo = albumIndexQuery.getPageNo();
        Integer pageSize = albumIndexQuery.getPageSize();
        albumSearchResponseVo.setPageNo(pageNo);
        albumSearchResponseVo.setPageSize(pageSize);

        //获取总记录数
        long total = searchResponse.hits().total().value();
        albumSearchResponseVo.setTotal(total);

        //计算总页数
        long totalPages = total % pageSize == 0 ? total / pageSize : total / pageSize + 1;
        albumSearchResponseVo.setTotalPages(totalPages);

        List<Hit<AlbumInfoIndex>> hits = searchResponse.hits().hits();

        if (CollUtil.isNotEmpty(hits)) {
            //用stream流将检索到的文档类型从AlbumInfoIndex 转换为 AlbumSearchResponseVo
            List<AlbumInfoIndexVo> list = hits.stream().map(hit -> {
                AlbumInfoIndex albumInfoIndex = hit.source();
                //处理高亮字段
                Map<String, List<String>> highlightMap = hit.highlight();
                if (CollUtil.isNotEmpty(highlightMap)) {
                    String highlightText = highlightMap.get("albumTitle").get(0);
                    albumInfoIndex.setAlbumTitle(highlightText);
                }

                return BeanUtil.copyProperties(albumInfoIndex, AlbumInfoIndexVo.class);
            }).collect(Collectors.toList());
            albumSearchResponseVo.setList(list);
        }




        return albumSearchResponseVo;
    }

    /**
     * 查询1级分类下置顶3级分类热度TOP6专辑
     *
     * @param category1Id
     * @return [{"baseCategory3":{三级分类对象},list:[专辑列表]},,{其他6个置顶分类热门专辑Map}]
     */
    @Override
    public List<Map<String, Object>> channel(Long category1Id) {
        try {
            //1.根据1级分类ID获取置顶三级分类ID
            //1.1 远程调用远程调用专辑微服务获取置顶3级分类
            List<BaseCategory3> baseCategory3List = albumFeignClient.findTopBaseCategory3(category1Id).getData();
            Assert.notNull(baseCategory3List, "一级分类{}置顶三级分类不存在", category1Id);

            //1.2 获取置顶三级分类ID列表
            List<FieldValue> fieldValueList = baseCategory3List
                    .stream()
                    .map(c3 -> FieldValue.of(c3.getId()))
                    .collect(Collectors.toList());

            //1.3 为了封装结果Map中分类对象，将三级分类List转为Map<3级分类ID，3级分类对象>
            Map<Long, BaseCategory3> category3Map = baseCategory3List
                    .stream()
                    .collect(Collectors.toMap(BaseCategory3::getId, baseCategory3 -> baseCategory3));

            //2. 调用ES进行检索：置顶分类热门专辑
            SearchResponse<AlbumInfoIndex> searchResponse = elasticsearchClient.search(
                    s -> s.index(INDEX_NAME)
                            .size(0)
                            .query(q -> q.terms(t -> t.field("category3Id").terms(t1 -> t1.value(fieldValueList))))
                            .aggregations(
                                    "c3_agg", a -> a.terms(t -> t.field("category3Id").size(7))
                                            .aggregations("top6", a1 -> a1.topHits(
                                                    t -> t.size(6)
                                                            .sort(s1 -> s1.field(f -> f.field("hotScore").order(SortOrder.Desc)))
                                                            .source(s1 -> s1.filter(f -> f.excludes("attributeValueIndexList",
                                                                    "hotScore",
                                                                    "commentStatNum",
                                                                    "buyStatNum",
                                                                    "subscribeStatNum",
                                                                    "announcerName")))
                                            ))
                            )
                    , AlbumInfoIndex.class);

            //3.解析聚合结果，获取置顶分类包含热度TOP6专辑列表 将7个置顶分类热门专辑Map封装到List集合中
            List<LongTermsBucket> longTermsBuckets = searchResponse.aggregations().get("c3_agg").lterms().buckets().array();
            //3.1 遍历三级分类聚合结果，每遍历一次就封装一个置顶分类热门专辑Map对象
            if (CollUtil.isNotEmpty(longTermsBuckets)) {
                List<Map<String, Object>> topCategory3MapList = longTermsBuckets
                        .stream()
                        .map(bucket -> {
                            //3.1.1 创建Map封装置顶分类热门专辑
                            Map<String, Object> map = new HashMap<>();
                            //3.1.2 获取聚合结果中三级分类ID
                            long topCategory3Id = bucket.key();
                            //3.1.3 获取top6专辑列表
                            //获取三级分类聚合内子聚合"top6"
                            List<Hit<JsonData>> top6HitsList = bucket.aggregations().get("top6")
                                    .topHits().hits().hits();
                            if (CollUtil.isNotEmpty(top6HitsList)) {
                                //遍历Hit集合 将集合泛型从Hit转为AlbumInfoIndex类型
                                List<AlbumInfoIndex> topAlbumList = top6HitsList.stream()
                                        .map(hit -> {
                                            //获取专辑JSON字符串
                                            String albumInfoIndexJSONStr = hit.source().toString();
                                            //将专辑JSON字符串转为AlbumInfoIndex对象
                                            AlbumInfoIndex albumInfoIndex = JSON.parseObject(albumInfoIndexJSONStr, AlbumInfoIndex.class);
                                            return albumInfoIndex;
                                        }).collect(Collectors.toList());
                                //3.2 封装置顶分类热门专辑Map
                                map.put("baseCategory3", category3Map.get(topCategory3Id));
                                map.put("list", topAlbumList);
                            }
                            return map;
                        }).collect(Collectors.toList());
                return topCategory3MapList;
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return null;
    }


    /**
     * 将专辑标题存入提词索引库
     *
     * @param albumInfoIndex
     */
    @Override
    public void saveSuggestIndex(AlbumInfoIndex albumInfoIndex) {
        //1.构建索引库文档对象
        SuggestIndex suggestIndex = new SuggestIndex();
        suggestIndex.setId(albumInfoIndex.getId().toString());
        String albumTitle = albumInfoIndex.getAlbumTitle();
        suggestIndex.setTitle(albumTitle);
        suggestIndex.setKeyword(new Completion(new String[]{albumTitle}));
        //1.1 将汉字转为汉语拼音 jing dian liu sheng ji
        String albumTitlePinyin = PinyinUtil.getPinyin(albumTitle, "");
        suggestIndex.setKeywordPinyin(new Completion(new String[]{albumTitlePinyin}));
        //1.1 将汉字转为汉语拼音首字母
        String albumTitleFirstLetter = PinyinUtil.getFirstLetter(albumTitle, "");
        suggestIndex.setKeywordSequence(new Completion(new String[]{albumTitleFirstLetter}));

        //2.存入提词文档记录到提词索引库
        suggestIndexRepository.save(suggestIndex);
    }


    private static final String SUGGEST_INDEX = "suggestinfo";

    /**
     * 根据用户已录入字符查询提词索引库进行自动补全关键字
     *
     * @param keyword
     * @return
     */
    @Override
    public List<String> completeSuggest(String keyword) {
        try {
            //1.发起自动补全请求
            SearchResponse<SuggestIndex> searchResponse = elasticsearchClient.search(s ->
                            s.index(SUGGEST_INDEX)
                                    .suggest(
                                            s1 -> s1.suggesters("letter-suggest", fs -> fs.prefix(keyword).completion(c -> c.field("keywordSequence").size(10).skipDuplicates(true)))
                                                    .suggesters("pinyin-suggest", s2 -> s2.prefix(keyword).completion(c -> c.field("keywordPinyin").size(10).skipDuplicates(true)))
                                                    .suggesters("keyword-suggest", s2 -> s2.prefix(keyword).completion(c -> c.field("keyword").size(10).skipDuplicates(true)))
                                    )
                    , SuggestIndex.class);
            //2.解析ES自动补全结果
            Set<String> hashSet = new HashSet<>();
            //2.1 解析建议结果-通过不同建议参数名获取汉字、拼音等提示词结果
            hashSet.addAll(this.parseSuggestResult(searchResponse, "letter-suggest"));
            hashSet.addAll(this.parseSuggestResult(searchResponse, "pinyin-suggest"));
            hashSet.addAll(this.parseSuggestResult(searchResponse, "keyword-suggest"));
            //2.2 如果解析建议提示词列表长度小于10，采用全文查询尝试补全到10个
            if (hashSet.size() < 10) {
                //2.2.1 根据用户录入字符进行全文检索
                SearchResponse<AlbumInfoIndex> matchSearchResponse = elasticsearchClient.search(
                        s -> s.index(INDEX_NAME)
                                .query(q -> q.match(m -> m.field("albumTitle").query(keyword)))
                                .size(10)
                        , AlbumInfoIndex.class
                );
                HitsMetadata<AlbumInfoIndex> hits = matchSearchResponse.hits();
                List<Hit<AlbumInfoIndex>> hitList = hits.hits();
                if (CollectionUtil.isNotEmpty(hitList)) {
                    for (Hit<AlbumInfoIndex> hit : hitList) {
                        AlbumInfoIndex source = hit.source();
                        //2.2.2 将检索到专辑标题内容加入到提词结果列表中
                        hashSet.add(source.getAlbumTitle());
                        if (hashSet.size() >= 10) {
                            break;
                        }
                    }
                }
            }
            if (hashSet.size() >= 10) {
                //如果提词结果列表中大于10截取前10个
                return new ArrayList<>(hashSet).subList(0, 10);
            } else {
                return new ArrayList<>(hashSet);
            }
        } catch (IOException e) {
            log.error("[搜索服务]关键字自动补全异常：{}", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 解析建议词结果
     *
     * @param searchResponse ES检索结果对象
     * @param suggestName    自定义建议词参数名称
     * @return
     */
    @Override
    public List<String> parseSuggestResult(SearchResponse<SuggestIndex> searchResponse, String suggestName) {
        //根据自定义建议词参数名称获取结果列表
        List<String> list = new ArrayList<>();
        List<Suggestion<SuggestIndex>> suggestionList = searchResponse.suggest().get(suggestName);
        if (CollectionUtil.isNotEmpty(suggestionList)) {
            //遍历得到建议对象
            for (Suggestion<SuggestIndex> suggestIndexSuggestion : suggestionList) {
                for (CompletionSuggestOption<SuggestIndex> option : suggestIndexSuggestion.completion().options()) {
                    SuggestIndex suggestIndex = option.source();
                    list.add(suggestIndex.getTitle());
                }
            }
        }
        return list;
    }

    @Override
    public Map<String, Object> albumInfo(Long albumId) {

        //0.TODO 采用布隆过滤器解决缓存穿透问题
        RBloomFilter<Object> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);

        boolean flag = bloomFilter.contains(albumId);

        if (!flag){
            throw new GuiguException(404,"专辑不存在");
        }

        ConcurrentHashMap<String, Object> map = new ConcurrentHashMap<>();

        //挨个远程调用

        //1.首先调用专辑信息接口
        CompletableFuture<AlbumInfo> albumInfoCompletableFuture = CompletableFuture.supplyAsync(() -> {
            log.info("调用专辑信息接口异步编排");
            AlbumInfo albumInfo = albumFeignClient.getAlbumInfo(albumId).getData();
            Assert.notNull(albumInfo, "专辑{}不存在", albumId);
            map.put("albumInfo", albumInfo);

            return albumInfo;
        }, threadPoolTaskExecutor);


        //2.由于需要用到之前调用接口的信息里的userId,所以不能先调user接口获取主播信息并封装
        CompletableFuture<Void> announcerCompletable = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            UserInfoVo announcer = userFeignClient.getUserInfoVo(albumInfo.getUserId()).getData();
            Assert.notNull(announcer, "用户{}不存在", announcer);
            map.put("announcer", announcer);
        }, threadPoolTaskExecutor);


        //3.调用专辑远程接口获取stat信息
        CompletableFuture<Void> albumStatVoCompletable = CompletableFuture.runAsync(() -> {
            AlbumStatVo albumStatVo = albumFeignClient.getAlbumStatVo(albumId).getData();
            Assert.notNull(albumStatVo, "专辑统计{}不存在", albumStatVo);
            map.put("albumStatVo", albumStatVo);
        }, threadPoolTaskExecutor);


        //4.调用远程接口获取 baseCategoryView
        CompletableFuture<Void> baseCategoryViewCompletable = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            BaseCategoryView baseCategoryView = albumFeignClient.getCategoryView(albumInfo.getCategory3Id()).getData();
            Assert.notNull(baseCategoryView, "专辑分类{}不存在", baseCategoryView);
            map.put("baseCategoryView", baseCategoryView);
        }, threadPoolTaskExecutor);

        //组合异步任务
        CompletableFuture.allOf(
                albumInfoCompletableFuture,
                announcerCompletable,
                albumStatVoCompletable,
                baseCategoryViewCompletable
        ).orTimeout(5, TimeUnit.SECONDS).join();

        return map;
    }

    @Override
    public void updateStatOnEs(TrackStatMqVo mqVo) {


        // 获取专辑ID
        Long albumId = mqVo.getAlbumId();


        AlbumInfoIndex albumInfoIndex = albumInfoIndexRepository.findById(albumId).orElse(null);

        if (albumInfoIndex == null) {
            return;
        }


        if (SystemConstant.TRACK_STAT_PLAY.equals(mqVo.getStatType())) {
            // 播放量在原有基础上增加
            albumInfoIndex.setPlayStatNum(albumInfoIndex.getPlayStatNum() + mqVo.getCount());
        }


        if (SystemConstant.TRACK_STAT_COMMENT.equals(mqVo.getStatType())) {
            // 评论量在原有基础上增加
            albumInfoIndex.setCommentStatNum(albumInfoIndex.getCommentStatNum() + mqVo.getCount());

        }


        albumInfoIndexRepository.save(albumInfoIndex);


    }


    /**
     * 更新Redis小时榜
     * @return
     */
    @Override
    public void updateLatelyAlbumRanking(Integer top) {

        try {
            //1.远程调用专辑服务获取所有的1级分类列表
            List<BaseCategory1> baseCategory1List = albumFeignClient.findAllCategory1().getData();


            Assert.notNull(baseCategory1List,"1级分类列表为空");

            for (BaseCategory1 baseCategory1 : baseCategory1List) {
                String[] rankingDimensionArray
                        = new String[]{"hotScore", "playStatNum", "subscribeStatNum", "buyStatNum", "commentStatNum"};

                //基于1级分类的id构建redis中hash结构的key
                String key = RedisConstant.RANKING_KEY_PREFIX + baseCategory1.getId();
                for (String dimension : rankingDimensionArray) {
                    //遍历5种排序,根据1级分类id＋排序字段获取专属TOPN记录
                    //检索ES

                    SearchResponse<AlbumInfoIndex> searchResponse = elasticsearchClient.search(
                            s -> s.index(INDEX_NAME)
                                    .query(q -> q.term(f -> f.field("category1Id").value(baseCategory1.getId())))
                                    .sort(s1 -> s1.field(f -> f.field(dimension).order(SortOrder.Desc)))
                                    .size(top)
                                    .source(s1 -> s1.filter(f -> f.excludes("attributeValueIndexList",
                                            "hotScore",
                                            "commentStatNum",
                                            "buyStatNum",
                                            "subscribeStatNum",
                                            "announcerName")))
                            , AlbumInfoIndex.class);

                    //解析ES种检索到的数据
                    List<Hit<AlbumInfoIndex>> hits = searchResponse.hits().hits();

                    if (CollUtil.isNotEmpty(hits)){
                        List<AlbumInfoIndex> topNList = hits.stream()
                                .map(hit -> hit.source()).collect(Collectors.toList());

                        //3.1 构建当前排序方式的小key
                        String field = dimension;

                        //将收集到的数据存入redis中
                        redisTemplate.opsForHash().put(key,field,topNList);
                    }

                }
            }
        } catch (IOException e) {
            log.error("更新redis小时榜数据异常:",e);
            throw new RuntimeException(e);
        }


    }

    /**
     * 获取不同分类下不同排行维度TOP20
     * @param category1Id
     * @param dimension
     * @return
     */
    @Override
    public List<AlbumInfoIndexVo> getRankingList(Long category1Id, String dimension) {

        //定义Redis中小时榜hash的Key
        String redisKey = RedisConstant.RANKING_KEY_PREFIX + category1Id;
        String field = dimension;

        //创建绑定hash对象方便操作hash结构

        BoundHashOperations<String,String,List<AlbumInfoIndex>> hashOps = redisTemplate.boundHashOps(redisKey);

        Boolean flag = hashOps.hasKey(field);

        if (flag){
            List<AlbumInfoIndex> list = hashOps.get(field);
            List<AlbumInfoIndexVo> albumInfoIndexVoList = list.stream()
                    .map(albumInfoIndex -> BeanUtil.copyProperties(albumInfoIndex, AlbumInfoIndexVo.class))
                    .collect(Collectors.toList());

            return albumInfoIndexVoList;
        }

        return null;
    }
}
